package com.chb.storm.kafkatopology;

import com.common.utils.PropUtils;

import backtype.storm.Config;
import backtype.storm.StormSubmitter;
import backtype.storm.generated.AlreadyAliveException;
import backtype.storm.generated.AuthorizationException;
import backtype.storm.generated.InvalidTopologyException;
import backtype.storm.spout.SchemeAsMultiScheme;
import backtype.storm.topology.TopologyBuilder;
import storm.kafka.KafkaSpout;
import storm.kafka.SpoutConfig;
import storm.kafka.StringScheme;
import storm.kafka.ZkHosts;

/**
 * 
 * gps历史表：
	create 'storm_ns:testKafkaToHbase', { NAME => 'd', COMPRESSION => 'snappy'}
	rowkey设计：word
	qualifier：content
 * @author Administrator
 *
 */
public class KafkaTopologyHbase {

	private static final String MYTOPIC_CONSUMER_SPOUT = "MYTOPIC_CONSUMER_SPOUT";
	
	private static final String LOG_PARSE_BOLT = "LOG_PARSE_BOLT";
	
	public static void main(String[] args) throws AlreadyAliveException, InvalidTopologyException, AuthorizationException {
		String brokerZkStr = PropUtils.getProp("brokerZkStr");
		String brokerZkPath = PropUtils.getProp("brokerZkPath");
		
		ZkHosts zkHosts = new ZkHosts(brokerZkStr, brokerZkPath);
		String topicName = "mytopic";
		
		//redeploy必须保证下面两个字段不变
		String zkRoot = "/kafka_zk";
		String spoutId = "KafkaTopologyHbase20021999";
		
		SpoutConfig kafkaConfig = new SpoutConfig(zkHosts, topicName, zkRoot, spoutId);
		kafkaConfig.scheme = new SchemeAsMultiScheme(new StringScheme());
		
		//忽略在zookeeper中的offset
		kafkaConfig.ignoreZkOffsets = true;
		//从当前部署时间来读取topic
		kafkaConfig.startOffsetTime = kafka.api.OffsetRequest.LatestTime();
		
		KafkaTopologyHbaseLogParseBolt logParseBolt = new KafkaTopologyHbaseLogParseBolt();
		
		TopologyBuilder builder = new TopologyBuilder();
		builder.setSpout(MYTOPIC_CONSUMER_SPOUT, new KafkaSpout(kafkaConfig), 1);
		builder.setBolt(LOG_PARSE_BOLT, logParseBolt, 1).shuffleGrouping(MYTOPIC_CONSUMER_SPOUT);
		
		Config config = new Config();
		config.setNumWorkers(3);
		StormSubmitter.submitTopology("KafkaTopologyHbase", config, builder.createTopology());
	}

}
